#pragma once
#ifdef WIN32
// Author: LiXizhi
// Date: 2010.5.11
// Desc: modified over boost/interprocess/ipc/message_queue.hpp to use windows_shared_memory.hpp.
// this allows us to use IPC even in low intergrity level process in vista and win7. 
#include <boost/interprocess/detail/config_begin.hpp>
#include <boost/interprocess/detail/workaround.hpp>

#include <boost/interprocess/windows_shared_memory.hpp>
#include <boost/interprocess/mapped_region.hpp>

#include <boost/interprocess/sync/interprocess_condition.hpp>
#include <boost/interprocess/sync/interprocess_mutex.hpp>
#include <boost/interprocess/detail/utilities.hpp>
#include <boost/interprocess/offset_ptr.hpp>
#include <boost/interprocess/creation_tags.hpp>
#include <boost/interprocess/exceptions.hpp>
#include <boost/detail/no_exceptions_support.hpp>
#include <boost/interprocess/detail/type_traits.hpp>

#include <algorithm> //std::lower_bound
#include <cstddef>   //std::size_t
#include <cstring>   //memcpy


//!\file
//!Describes an inter-process message queue. This class allows sending
//!messages between processes and allows blocking, non-blocking and timed
//!sending and receiving.

namespace ParaEngine{  namespace interprocess{
	using namespace boost;
	using namespace boost::interprocess;
	//!A class that allows sending messages
	//!between processes.
	class message_queue_win32
	{
		/// @cond
		//Blocking modes
		enum block_t   {  blocking,   timed,   non_blocking   };

		message_queue_win32();
		/// @endcond

	public:

		//!Creates a process shared message queue with name "name". For this message queue,
		//!the maximum number of messages will be "max_num_msg" and the maximum message size
		//!will be "max_msg_size". Throws on error and if the queue was previously created.
		message_queue_win32(create_only_t create_only,
			const char *name, 
			std::size_t max_num_msg, 
			std::size_t max_msg_size);

		//!Opens or creates a process shared message queue with name "name". 
		//!If the queue is created, the maximum number of messages will be "max_num_msg" 
		//!and the maximum message size will be "max_msg_size". If queue was previously 
		//!created the queue will be opened and "max_num_msg" and "max_msg_size" parameters
		//!are ignored. Throws on error.
		message_queue_win32(open_or_create_t open_or_create,
			const char *name, 
			std::size_t max_num_msg, 
			std::size_t max_msg_size);

		//!Opens a previously created process shared message queue with name "name". 
		//!If the was not previously created or there are no free resources, 
		//!throws an error.
		message_queue_win32(open_only_t open_only,
			const char *name);

		//!Destroys *this and indicates that the calling process is finished using
		//!the resource. All opened message queues are still
		//!valid after destruction. The destructor function will deallocate
		//!any system resources allocated by the system for use by this process for
		//!this resource. The resource can still be opened again calling
		//!the open constructor overload. To erase the message queue from the system
		//!use remove().
		~message_queue_win32(); 

		//!Sends a message stored in buffer "buffer" with size "buffer_size" in the 
		//!message queue with priority "priority". If the message queue is full
		//!the sender is blocked. Throws interprocess_error on error.*/
		void send (const void *buffer,     std::size_t buffer_size, 
			unsigned int priority);

		//!Sends a message stored in buffer "buffer" with size "buffer_size" through the 
		//!message queue with priority "priority". If the message queue is full
		//!the sender is not blocked and returns false, otherwise returns true.
		//!Throws interprocess_error on error.
		bool try_send    (const void *buffer,     std::size_t buffer_size, 
			unsigned int priority);

		//!Sends a message stored in buffer "buffer" with size "buffer_size" in the 
		//!message queue with priority "priority". If the message queue is full
		//!the sender retries until time "abs_time" is reached. Returns true if
		//!the message has been successfully sent. Returns false if timeout is reached.
		//!Throws interprocess_error on error.
		bool timed_send    (const void *buffer,     std::size_t buffer_size, 
			unsigned int priority,  const boost::posix_time::ptime& abs_time);

		//!Receives a message from the message queue. The message is stored in buffer 
		//!"buffer", which has size "buffer_size". The received message has size 
		//!"recvd_size" and priority "priority". If the message queue is empty
		//!the receiver is blocked. Throws interprocess_error on error.
		void receive (void *buffer,           std::size_t buffer_size, 
			std::size_t &recvd_size,unsigned int &priority);

		//!Receives a message from the message queue. The message is stored in buffer 
		//!"buffer", which has size "buffer_size". The received message has size 
		//!"recvd_size" and priority "priority". If the message queue is empty
		//!the receiver is not blocked and returns false, otherwise returns true.
		//!Throws interprocess_error on error.
		bool try_receive (void *buffer,           std::size_t buffer_size, 
			std::size_t &recvd_size,unsigned int &priority);

		//!Receives a message from the message queue. The message is stored in buffer 
		//!"buffer", which has size "buffer_size". The received message has size 
		//!"recvd_size" and priority "priority". If the message queue is empty
		//!the receiver retries until time "abs_time" is reached. Returns true if
		//!the message has been successfully sent. Returns false if timeout is reached.
		//!Throws interprocess_error on error.
		bool timed_receive (void *buffer,           std::size_t buffer_size, 
			std::size_t &recvd_size,unsigned int &priority,
			const boost::posix_time::ptime &abs_time);

		//!Returns the maximum number of messages allowed by the queue. The message
		//!queue must be opened or created previously. Otherwise, returns 0. 
		//!Never throws
		std::size_t get_max_msg() const;

		//!Returns the maximum size of message allowed by the queue. The message
		//!queue must be opened or created previously. Otherwise, returns 0. 
		//!Never throws
		std::size_t get_max_msg_size() const;

		//!Returns the number of messages currently stored. 
		//!Never throws
		std::size_t get_num_msg();

		//!Removes the message queue from the system.
		//!Returns false on error. Never throws
		static bool remove(const char *name);

		/** get the initial buffer pointer*/
		inline void* GetAddress() const {
			return m_pAddress;
		}
		/// @cond   
	private:
		typedef boost::posix_time::ptime ptime;
		bool do_receive(block_t block,
			void *buffer,            std::size_t buffer_size, 
			std::size_t &recvd_size, unsigned int &priority,
			const ptime &abs_time);

		bool do_send(block_t block,
			const void *buffer,      std::size_t buffer_size, 
			unsigned int priority,   const ptime &abs_time);

		//!Returns the needed memory size for the shared message queue.
		//!Never throws
		static std::size_t get_mem_size(std::size_t max_msg_size, std::size_t max_num_msg);

		boost::shared_ptr<windows_shared_memory> m_shmem;
		boost::shared_ptr<mapped_region> m_shmem_region;
		void * m_pAddress; 
		/// @endcond
	};

	/// @cond

	namespace detail {

		//!This header is the prefix of each message in the queue
		class msg_hdr_t 
		{
		public:
			std::size_t             len;     // Message length
			unsigned int            priority;// Message priority
			//!Returns the data buffer associated with this this message
			void * data(){ return this+1; }  //
		};
		// typedef msg_hdr msg_hdr_t;

		//!This functor is the predicate to order stored messages by priority
		class priority_functor
		{
		public:
			bool operator()(const offset_ptr<msg_hdr_t> &msg1, 
				const offset_ptr<msg_hdr_t> &msg2) const
			{  return msg1->priority < msg2->priority;  }
		};

		//!This header is placed in the beginning of the shared memory and contains 
		//!the data to control the queue. This class initializes the shared memory 
		//!in the following way: in ascending memory address with proper alignment
		//!fillings:
		//!
		//!-> mq_hdr_t: 
		//!   Main control block that controls the rest of the elements
		//!
		//!-> offset_ptr<msg_hdr_t> index [max_num_msg]
		//!   An array of pointers with size "max_num_msg" called index. Each pointer 
		//!   points to a preallocated message. The elements of this array are 
		//!   reordered in runtime in the following way:
		//!
		//!   When the current number of messages is "cur_num_msg", the first 
		//!   "cur_num_msg" pointers point to inserted messages and the rest
		//!   point to free messages. The first "cur_num_msg" pointers are
		//!   ordered by the priority of the pointed message and by insertion order 
		//!   if two messages have the same priority. So the next message to be 
		//!   used in a "receive" is pointed by index [cur_num_msg-1] and the first free
		//!   message ready to be used in a "send" operation is index [cur_num_msg].
		//!   This transforms index in a fixed size priority queue with an embedded free
		//!   message queue.
		//!
		//!-> struct message_t
		//!   {  
		//!      msg_hdr_t            header;
		//!      char[max_msg_size]   data;
		//!   } messages [max_num_msg];
		//!
		//!   An array of buffers of preallocated messages, each one prefixed with the
		//!   msg_hdr_t structure. Each of this message is pointed by one pointer of
		//!   the index structure.
		class mq_hdr_t
			: public detail::priority_functor
		{   
			typedef offset_ptr<msg_hdr_t> msg_hdr_ptr_t;
		public:
			//!Constructor. This object must be constructed in the beginning of the 
			//!shared memory of the size returned by the function "get_mem_size".
			//!This constructor initializes the needed resources and creates
			//!the internal structures like the priority index. This can throw.*/
			mq_hdr_t(std::size_t max_num_msg, std::size_t max_msg_size)
				: m_max_num_msg(max_num_msg), 
				m_max_msg_size(max_msg_size),
				m_cur_num_msg(0)
			{  this->initialize_memory();  }

			//!Returns the inserted message with top priority
			msg_hdr_t * top_msg()
			{  return mp_index[m_cur_num_msg-1].get();   }

			//!Returns true if the message queue is full
			bool is_full() const
			{  return m_cur_num_msg == m_max_num_msg;  }

			//!Returns true if the message queue is empty
			bool is_empty() const
			{  return !m_cur_num_msg;  }

			//!Frees the top priority message and saves it in the free message list
			void free_top_msg()
			{  --m_cur_num_msg;  }

			//!Returns the first free msg of the free message queue
			msg_hdr_t * free_msg()
			{  return mp_index[m_cur_num_msg].get();  }

			//!Inserts the first free message in the priority queue
			void queue_free_msg()
			{  
				//Get free msg
				msg_hdr_ptr_t free = mp_index[m_cur_num_msg];
				//Get priority queue's range
				msg_hdr_ptr_t *it  = &mp_index[0], *it_end = &mp_index[m_cur_num_msg];
				//Check where the free message should be placed
				it = std::lower_bound(it, it_end, free, static_cast<priority_functor&>(*this));
				//Make room in that position
				std::copy_backward(it, it_end, it_end+1);
				//Insert the free message in the correct position
				*it = free;
				++m_cur_num_msg;
			}

			//!Returns the number of bytes needed to construct a message queue with 
			//!"max_num_size" maximum number of messages and "max_msg_size" maximum 
			//!message size. Never throws.
			static std::size_t get_mem_size
				(std::size_t max_msg_size, std::size_t max_num_msg)
			{
				const std::size_t 
					msg_hdr_align  = boost::alignment_of<detail::msg_hdr_t>::value,
					index_align    = boost::alignment_of<msg_hdr_ptr_t>::value,
					r_hdr_size     = ipcdetail::ct_rounded_size<sizeof(mq_hdr_t), index_align>::value,
					r_index_size   = ipcdetail::get_rounded_size(sizeof(msg_hdr_ptr_t)*max_num_msg, msg_hdr_align),
					r_max_msg_size = ipcdetail::get_rounded_size(max_msg_size, msg_hdr_align) + sizeof(detail::msg_hdr_t);
				return r_hdr_size + r_index_size + (max_num_msg*r_max_msg_size);
			}

			//!Initializes the memory structures to preallocate messages and constructs the
			//!message index. Never throws.
			void initialize_memory()
			{
				const std::size_t 
					msg_hdr_align  = boost::alignment_of<detail::msg_hdr_t>::value,
					index_align    = boost::alignment_of<msg_hdr_ptr_t>::value,
					r_hdr_size     = ipcdetail::ct_rounded_size<sizeof(mq_hdr_t), index_align>::value,
					r_index_size   = ipcdetail::get_rounded_size(sizeof(msg_hdr_ptr_t)*m_max_num_msg, msg_hdr_align),
					r_max_msg_size = ipcdetail::get_rounded_size(m_max_msg_size, msg_hdr_align) + sizeof(detail::msg_hdr_t);

				//Pointer to the index
				msg_hdr_ptr_t *index =  reinterpret_cast<msg_hdr_ptr_t*>
					(reinterpret_cast<char*>(this)+r_hdr_size);

				//Pointer to the first message header
				detail::msg_hdr_t *msg_hdr   =  reinterpret_cast<detail::msg_hdr_t*>
					(reinterpret_cast<char*>(this)+r_hdr_size+r_index_size);  

				//Initialize the pointer to the index
				mp_index             = index;

				//Initialize the index so each slot points to a preallocated message
				for(std::size_t i = 0; i < m_max_num_msg; ++i){
					index[i] = msg_hdr;
					msg_hdr  = reinterpret_cast<detail::msg_hdr_t*>
						(reinterpret_cast<char*>(msg_hdr)+r_max_msg_size);
				}
			}

		public:
			//Pointer to the index
			offset_ptr<msg_hdr_ptr_t>  mp_index;
			//Maximum number of messages of the queue
			const std::size_t          m_max_num_msg;
			//Maximum size of messages of the queue
			const std::size_t          m_max_msg_size;
			//Current number of messages
			std::size_t                m_cur_num_msg;
			//Mutex to protect data structures
			interprocess_mutex         m_mutex;
			//Condition block receivers when there are no messages
			interprocess_condition     m_cond_recv;
			//Condition block senders when the queue is full
			interprocess_condition     m_cond_send;
		};


		//!This is the atomic functor to be executed when creating or opening 
		//!shared memory. Never throws
		class initialization_func_t
		{
		public:
			initialization_func_t(std::size_t maxmsg = 0, 
				std::size_t maxmsgsize = 0)
				: m_maxmsg (maxmsg), m_maxmsgsize(maxmsgsize) {}

			bool operator()(void *address, bool created)
			{
				char      *mptr;

				if(created){
					mptr     = reinterpret_cast<char*>(address);
					//Construct the message queue header at the beginning
					BOOST_TRY{
						new (mptr) mq_hdr_t(m_maxmsg, m_maxmsgsize);
					}
					BOOST_CATCH(...){
						return false;   
					}
					BOOST_CATCH_END
				}
				return true;
			}
			const std::size_t m_maxmsg;
			const std::size_t m_maxmsgsize;
		};

	}  //namespace detail {

	inline message_queue_win32::~message_queue_win32()
	{}

	inline std::size_t message_queue_win32::get_mem_size
		(std::size_t max_msg_size, std::size_t max_num_msg)
	{  return detail::mq_hdr_t::get_mem_size(max_msg_size, max_num_msg);   }

	inline message_queue_win32::message_queue_win32(create_only_t create_only,
		const char *name, 
		std::size_t max_num_msg, 
		std::size_t max_msg_size) : m_pAddress(NULL)
	{
		m_shmem.reset(new windows_shared_memory(create_only, name, read_write,get_mem_size(max_msg_size, max_num_msg)));
		m_shmem_region.reset(new mapped_region(*m_shmem, read_write));
		m_pAddress = m_shmem_region->get_address();

		// clear all memory
		detail::initialization_func_t clear_all(max_num_msg, max_msg_size);
		clear_all(GetAddress(), true);
	}

	inline message_queue_win32::message_queue_win32(open_or_create_t open_or_create,
		const char *name, 
		std::size_t max_num_msg, 
		std::size_t max_msg_size) : m_pAddress(NULL)
	{
		bool bCreated = false;
		try
		{
			// open first
			m_shmem.reset(new windows_shared_memory(open_only, name, read_write));
			bCreated = true;
		}
		catch (...)
		{
			// if not exist, open and create
			m_shmem.reset(new windows_shared_memory(open_or_create, name, read_write,get_mem_size(max_msg_size, max_num_msg)));
		}
		//Map the whole shared memory in this process
		m_shmem_region.reset(new mapped_region(*m_shmem, read_write));
		m_pAddress = m_shmem_region->get_address();

		if(!bCreated)
		{
			// clear all memory
			detail::initialization_func_t clear_all(max_num_msg, max_msg_size);
			clear_all(GetAddress(), true);
		}
	}

	inline message_queue_win32::message_queue_win32(open_only_t open_only,
		const char *name) : m_pAddress(NULL)
	{
		m_shmem.reset(new windows_shared_memory(open_only, name, read_write));
		m_shmem_region.reset(new mapped_region(*m_shmem, read_write));
		m_pAddress = m_shmem_region->get_address();
	}

	inline void message_queue_win32::send
		(const void *buffer, std::size_t buffer_size, unsigned int priority)
	{  this->do_send(blocking, buffer, buffer_size, priority, ptime()); }

	inline bool message_queue_win32::try_send
		(const void *buffer, std::size_t buffer_size, unsigned int priority)
	{  return this->do_send(non_blocking, buffer, buffer_size, priority, ptime()); }

	inline bool message_queue_win32::timed_send
		(const void *buffer, std::size_t buffer_size
		,unsigned int priority, const boost::posix_time::ptime &abs_time)
	{
		if(abs_time == boost::posix_time::pos_infin){
			this->send(buffer, buffer_size, priority);
			return true;
		}
		return this->do_send(timed, buffer, buffer_size, priority, abs_time);
	}

	inline bool message_queue_win32::do_send(block_t block,
		const void *buffer,      std::size_t buffer_size, 
		unsigned int priority,   const boost::posix_time::ptime &abs_time)
	{
		detail::mq_hdr_t *p_hdr = static_cast<detail::mq_hdr_t*>(GetAddress());
		//Check if buffer is smaller than maximum allowed
		if (buffer_size > p_hdr->m_max_msg_size) {
			throw interprocess_exception(size_error);
		}

		//---------------------------------------------
		scoped_lock<interprocess_mutex> lock(p_hdr->m_mutex);
		//---------------------------------------------
		{
			//If the queue is full execute blocking logic
			if (p_hdr->is_full()) {

				switch(block){
	case non_blocking :
		return false;
		break;

	case blocking :
		do{
			p_hdr->m_cond_send.wait(lock);
		}
		while (p_hdr->is_full());
		break;

	case timed :
		do{
			if(!p_hdr->m_cond_send.timed_wait(lock, abs_time)){
				if(p_hdr->is_full())
					return false;
				break;
			}
		}
		while (p_hdr->is_full());
		break;
	default:
		throw interprocess_exception(other_error);
				}
			}

			//Get the first free message from free message queue
			detail::msg_hdr_t *free_msg = p_hdr->free_msg();
			if (free_msg == 0) {
				throw interprocess_exception(other_error);
			}

			//Copy control data to the free message
			free_msg->priority = priority;
			free_msg->len      = buffer_size;

			//Copy user buffer to the message
			std::memcpy(free_msg->data(), buffer, buffer_size);

			//      bool was_empty = p_hdr->is_empty();
			//Insert the first free message in the priority queue
			p_hdr->queue_free_msg();

			//If this message changes the queue empty state, notify it to receivers
			//      if (was_empty){
			p_hdr->m_cond_recv.notify_one();
			//      }
		}  // Lock end

		return true;
	}

	inline void message_queue_win32::receive(void *buffer,              std::size_t buffer_size, 
		std::size_t &recvd_size,   unsigned int &priority)
	{  this->do_receive(blocking, buffer, buffer_size, recvd_size, priority, ptime()); }

	inline bool
		message_queue_win32::try_receive(void *buffer,              std::size_t buffer_size, 
		std::size_t &recvd_size,   unsigned int &priority)
	{  return this->do_receive(non_blocking, buffer, buffer_size, recvd_size, priority, ptime()); }

	inline bool
		message_queue_win32::timed_receive(void *buffer,              std::size_t buffer_size, 
		std::size_t &recvd_size,   unsigned int &priority,
		const boost::posix_time::ptime &abs_time)
	{
		if(abs_time == boost::posix_time::pos_infin){
			this->receive(buffer, buffer_size, recvd_size, priority);
			return true;
		}
		return this->do_receive(timed, buffer, buffer_size, recvd_size, priority, abs_time);
	}

	inline bool
		message_queue_win32::do_receive(block_t block,
		void *buffer,              std::size_t buffer_size, 
		std::size_t &recvd_size,   unsigned int &priority,
		const boost::posix_time::ptime &abs_time)
	{
		detail::mq_hdr_t *p_hdr = static_cast<detail::mq_hdr_t*>(GetAddress());
		//Check if buffer is big enough for any message
		if (buffer_size < p_hdr->m_max_msg_size) {
			throw interprocess_exception(size_error);
		}

		//---------------------------------------------
		scoped_lock<interprocess_mutex> lock(p_hdr->m_mutex);
		//---------------------------------------------
		{
			//If there are no messages execute blocking logic
			if (p_hdr->is_empty()) {
				switch(block){
	case non_blocking :
		return false;
		break;

	case blocking :
		do{
			p_hdr->m_cond_recv.wait(lock);
		}
		while (p_hdr->is_empty());
		break;

	case timed :
		do{
			if(!p_hdr->m_cond_recv.timed_wait(lock, abs_time)){
				if(p_hdr->is_empty())
					return false;
				break;
			}
		}
		while (p_hdr->is_empty());
		break;

		//Paranoia check
	default:
		throw interprocess_exception(other_error);
				}
			}

			//Thre is at least message ready to pick, get the top one
			detail::msg_hdr_t *top_msg = p_hdr->top_msg();

			//Paranoia check
			if (top_msg == 0) {
				throw interprocess_exception(other_error);
			}

			//Get data from the message
			recvd_size     = top_msg->len;
			priority       = top_msg->priority;

			//Copy data to receiver's bufers
			std::memcpy(buffer, top_msg->data(), recvd_size);

			//      bool was_full = p_hdr->is_full();

			//Free top message and put it in the free message list
			p_hdr->free_top_msg();

			//If this reception changes the queue full state, notify senders
			//      if (was_full){
			p_hdr->m_cond_send.notify_one();
			//      }
		}  //Lock end

		return true;
	}

	inline std::size_t message_queue_win32::get_max_msg() const
	{  
		detail::mq_hdr_t *p_hdr = static_cast<detail::mq_hdr_t*>(GetAddress());
		return p_hdr ? p_hdr->m_max_num_msg : 0;  }

	inline std::size_t message_queue_win32::get_max_msg_size() const
	{  
		detail::mq_hdr_t *p_hdr = static_cast<detail::mq_hdr_t*>(GetAddress());
		return p_hdr ? p_hdr->m_max_msg_size : 0;  
	}

	inline std::size_t message_queue_win32::get_num_msg()
	{  
		detail::mq_hdr_t *p_hdr = static_cast<detail::mq_hdr_t*>(GetAddress());
		if(p_hdr){
			//---------------------------------------------
			scoped_lock<interprocess_mutex> lock(p_hdr->m_mutex);
			//---------------------------------------------
			return p_hdr->m_cur_num_msg;
		}

		return 0;  
	}

	inline bool message_queue_win32::remove(const char *name)
	{  
		//return windows_shared_memory::remove(name);  
		return true;
	}

	/// @endcond

}} //namespace ParaEngine{  namespace interprocess{

#include <boost/interprocess/detail/config_end.hpp>
#endif